package com.shujia.flink.source;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class Demo3SourceFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //使用自定义source
        DataStream<Integer> myDS = env.addSource(new MySource());
        myDS.print();

        env.execute();
    }
}

//自定义source
//实现SourceFunction接口
class MySource implements SourceFunction<Integer> {
    //在run方法中读取外部的数据，使用原生java代码
    @Override
    public void run(SourceContext<Integer> ctx) throws Exception {
        while (true) {
            ctx.collect(1);
            Thread.sleep(1000);
        }
    }

    //cancel方法是任务被取消是执行的，用于回收资源
    @Override
    public void cancel() {
    }
}